package cn.engage.RocketMQ.Util;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @Auther: wwb
 * @Date:
 * @Description: 创建rocketmq的生产者
 */
@Component
@Slf4j
public class RocketMqProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送普通消息
     */
    public SendResult sendMsg(String topic, String tag, String keys, Object body) {
        return rocketMQTemplate.syncSend(topic + ":" + tag,
                MessageBuilder.withPayload(MqUtil.serialization(body)).setHeader(RocketMQHeaders.KEYS, keys).build());
    }

    /**
     * 发送普通消息
     */
    public SendResult sendMsg(String topic, String tag, Object body) {

        return rocketMQTemplate.syncSend(topic + ":" + tag,
                MessageBuilder.withPayload(MqUtil.serialization(body)).build());
    }

    /**
     * 发送普通消息
     */
    public SendResult sendMsg(String topic, Object body) {
        return rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(MqUtil.serialization(body)).build());
    }

    /**
     * 发送异步消息 在SendCallback中可处理相关成功失败时的逻辑
     * @param topic topic
     * @param tag tag
     * @param keys keys
     * @param body body
     * @param retryNum 重试次数
     */
    public void sendAsyncMsg(String topic, String tag, String keys, Object body, int retryNum) {
        asyncSend(topic, tag, keys, body, retryNum);
    }

    /**
     * 发送异步延时消息
     * @param topic
     * @param tag
     * @param keys
     * @param body
     * @param retryNum
     * @param delayLevel 延时级别 对应 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h
     * 2h
     */
    public void sendAsyncMsg(String topic, String tag, String keys, Object body, int retryNum, int delayLevel) {
        asyncSend(topic, tag, keys, body, retryNum, delayLevel);
    }

    /**
     * 发送异步消息 在SendCallback中可处理相关成功失败时的逻辑
     * @param topic
     * @param tag
     * @param body
     * @param retryNum 重试次数
     */
    public void sendAsyncMsg(String topic, String tag, Object body, int retryNum) {
        asyncSend(topic, tag, null, body, retryNum);
    }

    /**
     * 发送异步消息 在SendCallback中可处理相关成功失败时的逻辑
     * @param topic
     * @param body
     * @param retryNum 重试次数
     */
    public void sendAsyncMsg(String topic, Object body, int retryNum) {
        asyncSend(topic, null, null, body, retryNum);
    }

    private void asyncSend(String topic, String tag, String keys, Object body, int retryNum) {
        final String newTopic;
        if (StringUtils.isNotBlank(tag)) {
            newTopic = topic + ":" + tag;
        }
        else {
            newTopic = topic;
        }
        MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(MqUtil.serialization(body));
        if (StringUtils.isNotBlank(keys)) {
            stringMessageBuilder.setHeader(RocketMQHeaders.KEYS, keys);
        }

        rocketMQTemplate.asyncSend(newTopic, stringMessageBuilder.build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("Rocket处理成功：{},body:[{}]", sendResult, body);
                // 处理消息发送成功逻辑
            }

            @Override
            public void onException(Throwable e) {
                log.error("Rocket处理异常：", e);
                if (retryNum <= 0) {
                    return;
                }
                sendAsyncMsg(topic, tag, keys, body, retryNum - 1);
                log.debug("异步消息发送次数N= {}", retryNum);
            }
        });
    }

    /**
     * 发送延迟消息
     * @param topic
     * @param tag
     * @param keys
     * @param body
     * @param retryNum
     * @param delayLevel 延时级别 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    private void asyncSend(String topic, String tag, String keys, Object body, int retryNum, int delayLevel) {
        final String newTopic;
        if (StringUtils.isNotBlank(tag)) {
            newTopic = topic + ":" + tag;
        }
        else {
            newTopic = topic;
        }
        MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(MqUtil.serialization(body));
        if (StringUtils.isNotBlank(keys)) {
            stringMessageBuilder.setHeader(RocketMQHeaders.KEYS, keys);
        }

        rocketMQTemplate.asyncSend(newTopic, stringMessageBuilder.build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("Rocket处理成功：{},body:[{}]", sendResult, body);
                // 处理消息发送成功逻辑
            }

            @Override
            public void onException(Throwable e) {
                log.error("Rocket处理异常：", e);
                if (retryNum <= 0) {
                    return;
                }
                sendAsyncMsg(topic, tag, keys, body, retryNum - 1, delayLevel);
                log.debug("异步消息发送次数N= {}", retryNum);
            }
        }, rocketMQTemplate.getProducer().getSendMsgTimeout(), delayLevel);
    }


    /**
     * 发送延时消息<br/>
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h
     * 2h<br/>
     */
    public void sendDelayMsg(String msgBody, Integer delayLevel) {
        rocketMQTemplate.syncSend("spring_rocketMQ", MessageBuilder.withPayload(MqUtil.serialization(msgBody)).build(),
                30000000, delayLevel);
    }

//    public void  send() {
//        //  获取到这条信息
//        // 设置等级
//
//
//        /* 设置延迟消息级别  延迟推送是有等级之分的，百度 */
//        msg.setDelayTimeLevel(3);
//        // 发送消息到一个Broker
//        SendResult sendResult = producer.send(msg);
//
//        // 通过sendResult返回消息是否成功送达
//        System.out.printf(String.valueOf(sendResult));
//
//
//        // 发送消息  convertAndSend可以自动的将我们的对象转换为josn字符串
//       rocketMQTemplate.convertAndSend("spring_rocketMQ","group1", (Map<String, Object>) emp);
//
//    }
}
